package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.CartAddUserUnique;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Properties;

/**
 * @Author: LixuanFeng
 * @Date: 2022/9/1 19:17
 * @From: GoodGoodStudyDayDayUp
 *
 *
 * 需求： 对加购事实表 去重  求出人数
 *      状态中维护上次加购日期，如果状态为空 或者 当天日期和状态不一致 则 cartAddUuCt=1
 *
 *  先转换json对象，分组，状态去重，开窗聚合，即可
 *
 *  Doris 建表语句
 *
 *                drop table if exists dws_trade_cart_add_uu_window;
 *                create table if not exists dws_trade_cart_add_uu_window
 *                (
 *                    `stt`             DATETIME comment '窗口起始时间',
 *                    `edt`             DATETIME comment '窗口结束时间',
 *                    `cur_date`        DATE comment '当天日期',
 *                    `cart_add_uu_ct`  BIGINT replace comment '加购人数'
 *                ) engine = olap aggregate key (`stt`, `edt`, `cur_date`)
 *                comment "用户主题加购汇总表"
 *                partition by range(`cur_date`)()
 *                distributed by hash(`stt`) buckets 10 properties (
 *                  "replication_num" = "3",
 *                  "dynamic_partition.enable" = "true",
 *                  "dynamic_partition.time_unit" = "DAY",
 *                  "dynamic_partition.start" = "-1",
 *                  "dynamic_partition.end" = "1",
 *                  "dynamic_partition.prefix" = "par",
 *                  "dynamic_partition.buckets" = "10",
 *                  "dynamic_partition.hot_partition_num" = "1"
 *                );
 */
public class DwsUserCartAddUVWindow_lixuanfeng extends BaseAppOneStream {

    public static void main(String[] args) {

        new DwsUserCartAddUVWindow_lixuanfeng().init(3103,2,"DwsUserCartAddUVWindow_lixuanfeng",
                Constant.TOPIC_DWD_TRADE_CART_ADD);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        SingleOutputStreamOperator<CartAddUserUnique> reduce = stream
                .map(JSON::parseObject)
                // 找到某个用户的第一条加购记录
                .keyBy(obj -> obj.getString("user_id"))
                // 判定独立用户 , 状态维护 用户末次加购日期
                .process(new KeyedProcessFunction<String, JSONObject, CartAddUserUnique>() {

                    private ValueState<String> lastCartAddDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastCartAddDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastCartAddDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<CartAddUserUnique> out) throws Exception {

                        String lastCAdate = lastCartAddDate.value();
                        Long today = value.getLong("ts") * 1000;

                        String todayDate = AtguiguUtil.toDate(today);

                        Long cartAddUuCt = 0L;

                        //如果末次加购日期为 null 或者不等于当天日期，则保留数据并更新状态，否则丢弃
                        if (lastCAdate == null || !todayDate.equals(lastCAdate)) {


                            cartAddUuCt = 1L;
                            lastCartAddDate.update(todayDate);

                            out.collect(CartAddUserUnique.builder()
                                    .cartAddUuCt(cartAddUuCt)
                                    .ts(today)
                                    .build());
                        }
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<CartAddUserUnique>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean, ts) -> bean.getTs()))
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<CartAddUserUnique>() {
                            @Override
                            public CartAddUserUnique reduce(CartAddUserUnique value1,
                                                            CartAddUserUnique value2) throws Exception {
                                value1.setCartAddUuCt(value1.getCartAddUuCt() + value2.getCartAddUuCt());
                                return value1;
                            }
                        },
                        new ProcessAllWindowFunction<CartAddUserUnique, CartAddUserUnique, TimeWindow>() {
                            @Override
                            public void process(Context context, Iterable<CartAddUserUnique> elements,
                                                Collector<CartAddUserUnique> out) throws Exception {
                                CartAddUserUnique bean = elements.iterator().next();

                                String stt = AtguiguUtil.toDateTime(context.window().getStart());
                                String edt = AtguiguUtil.toDateTime(context.window().getEnd());
                                String cur = AtguiguUtil.toDate(System.currentTimeMillis());

                                bean.setStt(stt);
                                bean.setEdt(edt);
                                bean.setCurDate(cur);

                                out.collect(bean);
                            }
                        }
                );

     //   reduce.print("OKOKOKOK");

       Properties props = new Properties();
       props.setProperty("format", "json");
       props.setProperty("strip_outer_array", "true");
       String tableName= "edu.dws_trade_cart_add_uu_window";

           reduce
                   .map(bean ->{
                               SerializeConfig config = new SerializeConfig();
                               config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                               return JSON.toJSONString(bean, config);
                           }
                   )
                   .addSink(
                           DorisSink.sink(
                                   new DorisExecutionOptions.Builder()
                                           .setBatchIntervalMs(2000L)
                                           .setBatchSize(1024 * 1024)
                                           .setEnableDelete(false)
                                           .setMaxRetries(3)
                                           .setStreamLoadProp(props)
                                           .build(),
                                   new DorisOptions.Builder()
                                           .setFenodes(Constant.DORIS_HOST)
                                           .setUsername("root")
                                           .setPassword("aaaaaa")
                                           .setTableIdentifier(tableName)
                                           .build()));
    }
}
